package com.guguguweb.rpc.net.client.proxy;

import com.guguguweb.rpc.codec.MessageDecoder;
import com.guguguweb.rpc.codec.MessageEncoder;
import com.guguguweb.rpc.net.client.Handler.MessageClientHandler;
import com.guguguweb.rpc.net.client.imp.UnprocessedRequests;
import com.guguguweb.rpc.serializer.CommonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class ChannelProvider {

    private static final Logger logger = LoggerFactory.getLogger(ChannelProvider.class);

    private EventLoopGroup eventLoopGroup;

    private  Bootstrap bootstrap = initializeBootstrap();

    private  Map<InetSocketAddress, Channel> channels = new ConcurrentHashMap<>();


    private final UnprocessedRequests unprocessedRequests;

    public ChannelProvider(UnprocessedRequests unprocessedRequests){
        this.unprocessedRequests = unprocessedRequests;
    }

    private  Bootstrap initializeBootstrap() {
        eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                //连接的超时时间，超过这个时间还是建立不上的话则代表连接失败
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                //启用该功能时，TCP会主动探测空闲连接的有效性。可以将此功能视为TCP的心跳机制，默认的心跳间隔是7200s即2小时。
                .option(ChannelOption.SO_KEEPALIVE, true)
                //配置Channel参数，nodelay没有延迟，true就代表禁用Nagle算法，减小传输延迟。理解可参考：https://blog.csdn.net/lclwjl/article/details/80154565
                .option(ChannelOption.TCP_NODELAY, true);
        return bootstrap;
    }


    private ChannelProvider provider = this;


    /**
     * 复用channel,当连接第一次建立的时候会尝试发送连接，若连接成功则放入缓存当中
     * @param address
     * @param serializer
     * @return
     */
    public Channel get(InetSocketAddress address, CommonSerializer serializer) throws ExecutionException, InterruptedException, ConnectException {
        if(channels.containsKey(address)){
            Channel channel = channels.get(address);
            if(channel != null && channel.isActive()){
                return channel;
            }else {
                channels.remove(address);
            }
        }


        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch){
                ch.pipeline().addLast(new MessageEncoder(serializer))
                        //设定IdleStateHandler心跳检测每5秒进行一次写检测，如果5秒内write()方法未被调用则触发一次userEventTrigger()方法
                        //实现客户端每5秒向服务端发送一次消息
                        .addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS))
                        .addLast(new MessageDecoder())
                        .addLast(new MessageClientHandler(unprocessedRequests,provider));
            }
        });
        Channel channel = null;
        try {
            channel = connect(bootstrap, address);
        }
//        catch (ExecutionException | InterruptedException e){
//            logger.error("连接客户端时有错误发生", e);
//            throw e;
//        }
        catch (Exception e){
            logger.error("建立channel时出错，无法成功建立连接，地址：{}",address);
        }
        if(channel == null){
            throw new ConnectException("连接获得时channel失败，获得chanel为null");
        }
        channels.put(address,channel);
        return channel;
    }


    private static final Channel connectSyc(Bootstrap bootstrap, InetSocketAddress address){
        ChannelFuture connect = bootstrap.connect(address);
        return connect.channel();
    }


    private static final Channel connect(Bootstrap bootstrap, InetSocketAddress inetSocketAddress) throws
            ExecutionException, InterruptedException{
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                logger.info("客户端连接服务端成功,服务端 ：{}",inetSocketAddress);
                completableFuture.complete(future.channel());
            }else {
                completableFuture.complete(null);
                throw new IllegalStateException();
            }
        });
        return completableFuture.get();
    }
}
